package com.atguigu.bigdata.chapter07.watermark;

import com.atguigu.bigdata.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;

/**
 * @Author lzc
 * @Date 2022/9/4 8:28
 */
public class Flink04_WMK_AllowLateness {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        // 今天测试水印的时候: 并行度必须设置 1
        env.setParallelism(1);
        
        // 抽象时间时间
        // 返回值就是这个与元素的事件时间
        env
            .socketTextStream("hadoop162", 9999)
            .map(new MapFunction<String, WaterSensor>() {
                @Override
                public WaterSensor map(String value) throws Exception {
                    String[] data = value.split(",");
                    return new WaterSensor(data[0], Long.valueOf(data[1]), Integer.valueOf(data[2]));
                }
            })
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    // 返回值必须是毫秒. 是一个绝对时间
                    .withTimestampAssigner((element, ts) -> element.getTs())
            )
            .keyBy(WaterSensor::getId)
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            // 允许迟到2s
            .allowedLateness(Time.seconds(2))
            /* .process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
                 
                 // 处理窗口中元素的方法: 当窗口关闭的时候执行一次
                 @Override
                 public void process(String key,  // 属于哪个key的窗口
                                     Context ctx,  // 上下文对象, 可以获取到窗口信息
                                     Iterable<WaterSensor> elements, // 窗口触发计算的时候, 存储着窗口内所有的元素
                                     Collector<String> out) throws Exception {
                     ArrayList<WaterSensor> list = new ArrayList<>();
                     // Iterable很多操作不能做, 比如排序, 一般是把他转成一个List集合再操作
                     for (WaterSensor ws : elements) {
                         list.add(ws);
                     }
                     
                     // 获取窗口的开始时间
                     String stt = AtguiguUtil.toDatTime(ctx.window().getStart());
                     // 获取窗口的结束时间
                     String edt = AtguiguUtil.toDatTime(ctx.window().getEnd());
                     
                     out.collect(key + " " + stt + " " + edt + "  " + list);
                 }
             })*/
            .sum("vc")
            .print();
        
        
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
/*
1. 解决数据迟到问题的第一重保证
        事件时间+水印

2. 解决数据迟到的第二重保证
        允许迟到
        
        当到了窗口关闭时间, 窗口先不关闭, 只是对窗口内的元素进行计算.
        然后允许迟到期间, 如果有属于这个窗口的数据来了, 则来一条就计算一次.
        超过了允许迟到时间, 窗口才真正的关闭!!!!
        
3. 解决数据迟到的第三重保证
    测输出路
        如果窗口真正关闭之后, 这是时候来的属于关闭窗口的数据, 会直接进入侧输出流
*/